package com.hydra.cron.common.curator.helper;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hydra.cron.common.curator.selector.LeaderEngine;
import com.hydra.cron.common.curator.selector.RunningListener;
import com.hydra.cron.common.utils.ConfigUtil;
import com.hydra.cron.common.utils.ObjectUtil;
import com.hydra.cron.common.utils.StrUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by lee on 15/11/30.
 * 获取client对象
 */
public class CuratorHelper {
    private static final String CONNECT_STRING_NAME = "zookeeper.connectString";
    private static final String SESSION_TIMEOUT_NAME = "zookeeper.sessionTimeout";
    private static final String CONNECTION_TIMEOUT_NAME = "zookeeper.connectionTimeout";
    private static final String NAME_SPACE = "zookeeper.namespace";

    private static final Logger LOG = LoggerFactory.getLogger(CuratorHelper.class);

    private static final ExecutorService exec = Executors.newFixedThreadPool(4);
//    private static List<NodeCache> nodeCaches = Lists.newCopyOnWriteArrayList();

    private Map<String, NodeCache> watchNodeMap = Maps.newConcurrentMap();
    private Map<String, PathChildrenCache> watchChildMap = Maps.newConcurrentMap();

    private volatile static CuratorHelper instance = null;
    private volatile static LeaderEngine leaderEngine = null;

    private CuratorFramework client;

    private CuratorHelper() {

    }

    private void init() {
        Map<String, String> properties = ConfigUtil.getConfigsByFile("zookeeper", "zookeeper");
        String connectString = properties.get(CONNECT_STRING_NAME);
        String sessionTimeoutString = properties.get(SESSION_TIMEOUT_NAME);
        String connectionTimeoutString = properties.get(CONNECTION_TIMEOUT_NAME);
        String namespace = properties.get(NAME_SPACE);
        if (StrUtil.isBlank(connectString)) {
            throw new RuntimeException("zookeeper properties is error");
        }
        client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(Integer.parseInt(sessionTimeoutString))
                .connectionTimeoutMs(Integer.parseInt(connectionTimeoutString))
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace(namespace)
                .build();
        client.start();
    }

    public CuratorFramework getClient() {
        boolean established = false;
        try {
            established = client.blockUntilConnected(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        if (established) {
            return client;
        }
        CloseableUtils.closeQuietly(client);
        return null;
    }

    public static CuratorHelper getInstance() {
        if (ObjectUtil.isNotNull(instance)) {
            return instance;
        } else {
            synchronized (CuratorHelper.class) {
                if (ObjectUtil.isNull(instance)) {
                    CuratorHelper c = new CuratorHelper();
                    c.init();
                    instance = c;
                }
            }
        }

        return instance;
    }

    public boolean checkExists(String znode) throws Exception {
        return ObjectUtil.isNotNull(CuratorHelper.getInstance().getClient().checkExists().forPath(znode));
    }

    public void create(String znode, String value) throws Exception {
        getInstance().getClient().create().creatingParentsIfNeeded().forPath(znode, value.getBytes());
    }

    public void createEphemeral(String znode, String value) throws Exception {
        getInstance().getClient().create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(znode, value.getBytes());
    }

    /**
     * 获取节点内容
     *
     * @param znode
     * @return
     * @throws Exception
     */
    public String getData(String znode) throws Exception {
        if (checkExists(znode)) {
            return new String(CuratorHelper.getInstance().getClient().getData().forPath(znode), Charset.forName("UTF-8"));
        }
        return null;
    }

    /**
     * 设置节点值
     *
     * @param znode
     * @param value
     * @throws Exception
     */
    public void update(String znode, String value) throws Exception {
        client.setData().forPath(znode, value.getBytes());
    }

    /**
     * 删除节点
     *
     * @param znode
     * @throws Exception
     */
    public void delete(String znode) throws Exception {
        if (checkExists(znode)) {
            CuratorHelper.getInstance().getClient().delete().forPath(znode);
        }
    }

    /**
     * 获取所有子节点
     *
     * @param path
     * @return
     * @throws Exception
     */
    public static List<String> getChildrenNodePath(String path) throws Exception {
        try {
            return getInstance().getClient().getChildren().forPath(path);
        } catch (Exception e) {
            LOG.error("get Error", e);
        }
        return Lists.newArrayList();
    }

    /**
     * 监听节点
     *
     * @param path
     * @param listener
     * @throws Exception
     */
    public void watch(String path, NodeCacheListener listener) throws Exception {
        NodeCache nodeCache = new NodeCache(getInstance().getClient(), path, false);
        nodeCache.getListenable().addListener(listener, exec);
        nodeCache.start(true);
        watchNodeMap.put(path, nodeCache);
    }

    /**
     * 监听节点下所有子节点变化
     *
     * @param path
     * @param listener
     * @throws Exception
     */
    public void watchChild(String path, PathChildrenCacheListener listener) throws Exception {
        PathChildrenCache childrenCache = new PathChildrenCache(getInstance().getClient(), path, true);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        childrenCache.getListenable().addListener(listener, exec);
        childrenCache.start();
        watchChildMap.put(path, childrenCache);
    }

    public void removeWatch(String path) throws IOException {
        NodeCache nodeCache = watchNodeMap.get(path);

        if (ObjectUtil.isNotNull(nodeCache)) {
            nodeCache.close();
            watchNodeMap.remove(path);
        }
    }

    public void removeWatchChild(String path) throws IOException {
        PathChildrenCache nodeCache = watchChildMap.get(path);

        if (ObjectUtil.isNotNull(nodeCache)) {
            nodeCache.close();
            watchChildMap.remove(path);
        }
    }

    /**
     * 关闭nodeCache
     *
     * @throws IOException
     */
    public void closeNodeCache() throws IOException {
        for (NodeCache nodeCache : watchNodeMap.values()) {
            nodeCache.close();
        }
        watchNodeMap.clear();

        for (PathChildrenCache nodeCache : watchChildMap.values()) {
            nodeCache.close();
        }
        watchChildMap.clear();
    }

    public void takeLeader(String path, RunningListener listener) throws IOException {
        if (ObjectUtil.isNull(leaderEngine)) {
            synchronized (CuratorHelper.class) {
                if (ObjectUtil.isNull(leaderEngine)) {
                    LeaderEngine l = new LeaderEngine(getInstance().getClient(), path);
                    l.setListener(listener);
                    l.start();
                    leaderEngine = l;
                }
            }
        } else {
            throw new RuntimeException("leader is exist");
        }
    }

    public boolean isLeader() throws IOException {
        return ObjectUtil.isNotNull(leaderEngine) && leaderEngine.isLeader();
    }

    public void stopTakeLeader() throws IOException {
        if (ObjectUtil.isNotNull(leaderEngine)) {
            leaderEngine.close();
            leaderEngine = null;
        } else {
            throw new RuntimeException("leader is not exist");
        }
    }
}
